package javaVersion.learn.stateProcess.operatorStateDemo;

import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.runtime.state.FunctionInitializationContext;
import org.apache.flink.runtime.state.FunctionSnapshotContext;
import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;

import java.util.Iterator;

/**
 * @program: myFlink
 * @description: 自定义数据源、自动产生数据、手动抛异常、保存数据偏移量到ck
 * @author: WincoMa
 * @create: 2020-07-27 11:59
 **/
public class MyRichSource extends RichSourceFunction<String> implements CheckpointedFunction {
    private Boolean isRunning = true;

    private ListState<Long> offsetState = null;

    //offset偏移量
    private Long offset = 0L;

    @Override
    public void run(SourceContext<String> ctx) throws Exception {
        Iterator<Long> iterator = offsetState.get().iterator();
        if (iterator.hasNext()) offset = iterator.next();

        //循环生成数据
        while (isRunning) {
            offset += 1;
            ctx.collect("offset->" + offset);
            Thread.sleep(1000);
            //如果offset=10，手动抛出异常
            if (offset == 10) {
                System.out.println(1 / 0);
            }
        }
    }

    @Override
    public void cancel() {
        isRunning = false;
    }

    @Override
    public void snapshotState(FunctionSnapshotContext context) throws Exception {
        //首先将原offset清空
        offsetState.clear();
        //然后将新offset添加,做快照
        offsetState.add(offset);
    }

    @Override
    public void initializeState(FunctionInitializationContext context) throws Exception {
        //使用state描述器获取状态

        //这里使用了类型擦除，TypeHint按时数据类型
        ListStateDescriptor<Long> listStateDescriptor = new ListStateDescriptor<>("offset", TypeInformation.of(
                new TypeHint<Long>() {
                }
        ));

        //上下文对象中获取指定的state（状态信息）
        offsetState = context.getOperatorStateStore().getListState(listStateDescriptor);


    }
}
